package com.bw.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.bean.TansportTask;
import com.bw.util.MyKafkaUtil;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;

import java.text.SimpleDateFormat;
import java.util.Properties;
import java.util.TimeZone;

public class Test4 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);


        DataStream<String> stream = env.addSource(MyKafkaUtil.getKafkaConsumer("tms_ods_yk6","test668"));

        //（4）实时数仓DWD层-业务从原始业务数据库的transport_task表中提取运单完成操作，写入Kafka指定主题。
        // 环境准备，从 Kafka tms_ods 主题读取数据，筛选运输完成数据，处理运输完成数据，获取修改后的数据并转换数据类型，补全时间戳字段，
        // 补全运输时长字段，处理时区问题，脱敏，写出到 Kafka tms_dwd_trans_trans_finish 主题物流域运输完成事实主题（5分）
        // 67004 代表的已完成
//        SingleOutputStreamOperator<String> ttDS = stream.filter(x -> {
//            return JSON.parseObject(x).getString("table").equals("transport_task") &&
//                    JSON.parseObject(x).getJSONObject("data").getString("status").equals("67004");
//        });

        SingleOutputStreamOperator<String> ttDS = stream.filter(new FilterFunction<String>() {
            @Override
            public boolean filter(String s) throws Exception {
                try {
                    JSONObject jsonObject = JSON.parseObject(s);
                    String table = jsonObject.getString("table");
                    String status = jsonObject.getJSONObject("data").getString("status");
                    if ("transport_task".equals(table) && "67004".equals(status)) {
                        return true;
                    }
                } catch (Exception e) {
//                    e.printStackTrace();
                    return false;
                }
                return false;
            }
        });

        ttDS.print();
        //{"database":"tms01","data":{"end_org_name":"湖北省武汉市转运中心","truck_id":945,"driver2_name":"凤山","create_time":"2023-07-12T21:50:16Z","driver1_emp_id":1712,"start_org_id":15,"driver2_emp_id":1713,"line_id":857,"start_org_name":"江西省南昌市转运中心","driver1_name":"司空风航","update_time":"2023-07-13T03:50:59Z","is_deleted":"0","shift_id":1890,"actual_distance":344.00,"actual_end_time":1689202259000,"id":105849,"order_num":1,"truck_no":"赣DGD48Z","actual_start_time":1689184823000,"end_org_id":19,"status":"67004"},"type":"insert","table":"transport_task"}
        SingleOutputStreamOperator<String> map = ttDS.map(new MapFunction<String, String>() {
            @Override
            public String map(String s) throws Exception {
                //补全时间戳字段  补全运输时长字段  处理时区问题  脱敏（快递员姓名或其他？）
                //,"create_time":"2023-07-12T21:50:16Z"
                //"driver2_name":"凤山"
                //"actual_end_time":1689202259000,
                TansportTask tt = JSON.parseObject(JSON.parseObject(s).getString("data"), TansportTask.class);
                SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss'Z'");
                // UTC
                sdf.setTimeZone(TimeZone.getTimeZone("UTC"));


                // 获取时间戳
                long time = sdf.parse(tt.getCreate_time()).getTime();
                tt.setCreate_ts(time);

                // 运输时长
                tt.setTransport_time(Long.parseLong(tt.getActual_end_time()) - Long.parseLong(tt.getActual_start_time()));
                // 脱敏
                tt.setDriver1_name(tt.getDriver1_name() != null ? (tt.getDriver1_name().substring(0, 1) + "**") : tt.getDriver1_name());
                tt.setDriver2_name(tt.getDriver1_name() != null ? (tt.getDriver1_name().substring(0, 1) + "**") : tt.getDriver1_name());

                return JSON.toJSONString(tt);
            }
        });

        map.print();
        //{"actual_distance":225.0,"actual_end_time":"1689196253000","actual_start_time":"1689184823000","create_time":"2023-07-12T21:50:16Z","create_ts":1689198616000,"driver1_emp_id":1656,"driver1_name":"张**","driver2_name":"张**","end_org_id":31,"end_org_name":"青海省西宁市转运中心","id":105834,"is_deleted":"0","line_id":844,"order_num":0,"shift_id":1830,"start_org_id":30,"start_org_name":"甘肃省兰州市转运中心","status":"67004","transport_time":11430000,"truck_id":915,"truck_no":"青AGDBHL","update_time":"2023-07-13T02:10:53Z"}
        //保存到：tms_dwd_trans_trans_finish
        map.addSink(MyKafkaUtil.getKafkaProducer("tms_dwd_trans_trans_finish_yk6"));

        env.execute();
    }
}
